package com.sunday.common.mq.rabbit.study.spring.study;


import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.config.ContainerCustomizer;
import org.springframework.amqp.rabbit.config.SimpleRabbitListenerContainerFactory;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
import org.springframework.amqp.rabbit.listener.api.RabbitListenerErrorHandler;
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.beans.factory.ObjectProvider;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.autoconfigure.amqp.RabbitTemplateConfigurer;
import org.springframework.boot.autoconfigure.amqp.RabbitTemplateCustomizer;
import org.springframework.boot.autoconfigure.amqp.SimpleRabbitListenerContainerFactoryConfigurer;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.retry.backoff.ExponentialBackOffPolicy;
import org.springframework.retry.support.RetryTemplate;

/**
 * org.springframework.boot.autoconfigure.amqp.RabbitAutoConfiguration
 * https://github.com/openzipkin/brave/blob/master/instrumentation/spring-rabbit/README.md
 * <p>
 * spring-amqp二个核心类RabbitAdmin和RabbitTemplate类
 * 1.RabbitAdmin类完成对Exchange，Queue，Binging的操作，在容器中管理了RabbitAdmin类的时候，可以对Exchange，Queue，Binging进行自动声明。
 * 2.RabbitTemplate类是发送和接收消息的工具类。
 * <p>
 * RabbitTemplate类是简化RabbitMQ访问的工具类（发送和接收消息）
 * 1.使用RabbitTemplate进行消息的发送。
 * 2.使用SimpleMessageListenerContainer类监听队列，进行消息的消费。
 */
@Slf4j
@Configuration
public class EmptyRabbitBraveConfiguration {

    @Value("${spring.application.name}")
    private String applicationName;

    @Bean
    public Jackson2JsonMessageConverter jackson2JsonMessageConverter() {
        return new Jackson2JsonMessageConverter();
    }

    @Bean
    public RabbitTemplate rabbitTemplate(RabbitTemplateConfigurer configurer, ConnectionFactory connectionFactory, ObjectProvider<RabbitTemplateCustomizer> customizers) throws Exception {
        RabbitTemplate template = new RabbitTemplate();
        configurer.configure(template, connectionFactory);
        customizers.orderedStream().forEach((customizer) -> customizer.customize(template));
        template.setConfirmCallback((correlationData, ack, cause) -> {
            log.info("correlationData : {}", correlationData);
            log.info("ack : {}", ack);
            log.info("cause : {}", cause);
        });
        template.setReturnsCallback(returned -> {
            log.info("exchange : {}", returned.getExchange());
            log.info("routingKey : {}", returned.getRoutingKey());
            log.info("message : {}", returned.getMessage());
            log.info("replyCode : {}", returned.getReplyCode());
            log.info("replyText : {}", returned.getReplyText());
        });
        template.setMessageConverter(jackson2JsonMessageConverter());
        return template;
    }

    @Bean
    public RetryTemplate retryTemplate(RabbitTemplate rabbitTemplate) throws Exception {
        RetryTemplate retryTemplate = new RetryTemplate();
        ExponentialBackOffPolicy backOffPolicy = new ExponentialBackOffPolicy();
        backOffPolicy.setInitialInterval(500);
        backOffPolicy.setMultiplier(10.0);
        backOffPolicy.setMaxInterval(10000);
        retryTemplate.setBackOffPolicy(backOffPolicy);
        return retryTemplate;
    }

    @Bean
    public SimpleRabbitListenerContainerFactory rabbitListenerContainerFactory(
            SimpleRabbitListenerContainerFactoryConfigurer configurer,
            ConnectionFactory connectionFactory,
            ObjectProvider<ContainerCustomizer<SimpleMessageListenerContainer>> simpleContainerCustomizer
    ) {
        SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
        configurer.configure(factory, connectionFactory);
        simpleContainerCustomizer.ifUnique(factory::setContainerCustomizer);
        /**
         * 预取默认值曾经是 1，这可能导致高效使用者的利用率不足。 从版本 2.0 开始，默认预取值现在为 250，这应该让使用者在最常见的场景中保持忙碌和 从而提高吞吐量。
         * 尽管如此，在某些情况下，预取值应该很低：
         * 对于大型消息，尤其是在处理速度较慢的情况下（消息可能会在客户端进程中增加大量内存）
         * 当需要严格的消息排序时（在这种情况下，预取值应设置回 1）
         * 其他特殊情况
         * 此外，对于低容量消息传递和多个使用者（包括单个侦听器容器实例中的并发），您可能希望减少预取，以便在使用者之间更均匀地分配消息。
         */
        factory.setPrefetchCount(1);

        factory.setMessageConverter(jackson2JsonMessageConverter());

        factory.setConsumerTagStrategy(queue -> applicationName + "_" + queue);

        return factory;
    }

    @Bean
    public RabbitListenerErrorHandler rabbitListenerErrorHandler() {
        return (amqpMessage, message, exception) -> {
            log.info("Message : {}", amqpMessage);
            log.info("org.springframework.messaging.Message : {}", message);
            log.info("ListenerExecutionFailedException : {}", exception);
            return null;
        };
    }


    /**
     * https://docs.spring.io/spring-amqp/docs/current/reference/html/#receiving-batch
     *
     * @RabbitListener批处理 接收一批消息时，通常由容器执行去批处理，并且一次使用一条消息调用侦听器。
     * 从版本 2.2 开始，您可以将侦听器容器工厂和侦听器配置为在一次调用中接收整个批处理，只需设置工厂的属性，并使方法有效负载参数为 a 或 ：batchListenerListCollection
     * <p>
     * 同样从 2.2 版开始。一次接收一条批处理消息时，最后一条消息包含设置为 的布尔标头。 可以通过将布尔 last' 参数添加到侦听器方法来获取此标头。
     * 标头映射自 。 此外，填充了每个消息片段中的批处理大小。true @Header(AmqpHeaders.LAST_IN_BATCH) MessageProperties.isLastInBatch() AmqpHeaders.BATCH_SIZE
     * 此外，还向 中添加了一个新属性。 如果为 true，容器将创建一批消息，最多 ;如果经过没有新消息到达，则会传递部分批处理。 如果收到生产者创建的批处理，则会对其进行去批处理并将其添加到使用者端批处理中;因此，实际传送的消息数可能超过 ，这表示从代理接收的消息数。 当为真时必须为真;容器工厂将强制实施此要求。consumerBatchEnabledSimpleMessageListenerContainerbatchSizereceiveTimeoutbatchSizedeBatchingEnabledconsumerBatchEnabled
     */
    @Bean
    public SimpleRabbitListenerContainerFactory consumerBatchContainerFactory(
            SimpleRabbitListenerContainerFactoryConfigurer configurer,
            ConnectionFactory connectionFactory,
            ObjectProvider<ContainerCustomizer<SimpleMessageListenerContainer>> simpleContainerCustomizer
    ) {
        SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
        configurer.configure(factory, connectionFactory);
        simpleContainerCustomizer.ifUnique(factory::setContainerCustomizer);

        factory.setConsumerTagStrategy(queue -> applicationName + "_" + queue);

//        factory.setPrefetchCount(1);
        factory.setBatchSize(2);
        // 同时也将设置 setBatchListener(true);
        factory.setConsumerBatchEnabled(true);

        return factory;
    }

}
